I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

Draft stacked on #20143.

• Add an Iceberg-compatible bucket(count, value) partition transform.
• Add WRITE_DISTRIBUTION_MODE = hash for partitioned Fuse tables.
• For eligible INSERT SELECT plans, compose the existing EvalScalar -> GlobalHash -> Sort physical operators before DistributedInsertSelect.
• Reuse the existing Fuse append, compact, partition split, serialization, and commit paths without adding new execution processors.

The goal is to reduce fragmented blocks when unordered input repeatedly crosses physical partition boundaries.

Current scope

• Hash write distribution currently optimizes INSERT SELECT when no schema cast is required and all partition source columns are present.
• Other INSERT sources and unsupported INSERT SELECT shapes fall back to the existing append path.
• bucket(count, value) controls physical partition cardinality. It does not configure the number of execution nodes or pipeline lanes.
• This draft intentionally keeps the execution change small while a common physical table-write sink remains future work.

Tests

• Unit Test
  • CARGO_BUILD_JOBS=2 env REGENERATE_GOLDENFILES=1 cargo test -p databend-common-functions --test it test_hash
  • CARGO_BUILD_JOBS=2 cargo check -p databend-query
• Logic Test - coverage added in 09_0054_partition_by.test, execution pending
• Benchmark Test
• No Test - Pair with the reviewer to explain why

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

This change is